package com.github.yoojia.mqtt;

import com.github.yoojia.mqtt.util.SnowflakeId;
import com.parkingwang.lang.data.Pair;
import com.parkingwang.lang.kit.StringKit;
import org.eclipse.paho.client.mqttv3.IMqttMessageListener;
import org.eclipse.paho.client.mqttv3.MqttClient;
import org.eclipse.paho.client.mqttv3.MqttException;
import org.eclipse.paho.client.mqttv3.MqttMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;

/**
 * @author Yoojia Chen (yoojiachen@gmail.com)
 * @since 1.0.0
 */
final class Subscriber {

    public static final Logger LOGGER = LoggerFactory.getLogger(Subscriber.class);

    private final Map<Long, Pair<LocalFilter, MessageHandler>> mArgsMap = new ConcurrentHashMap<>();

    private final MQTTSocket mSocket;
    private final SnowflakeId mRegSerialId;
    private final MqttClient mClient;
    private final String mTopicFilter;

    public Subscriber(MQTTSocket socket, SnowflakeId snowflakeId, MqttClient client, String topicFilter) {
        mSocket = socket;
        mRegSerialId = snowflakeId;
        mClient = client;
        mTopicFilter = topicFilter;
    }

    public String getTopicFilter() {
        return mTopicFilter;
    }

    public void subscribe() throws MqttException {
        LOGGER.info("Subscribing with filter topic: {}", mTopicFilter);
        mClient.subscribe(mTopicFilter, new IMqttMessageListener() {
            @Override
            public void messageArrived(String topic, MqttMessage mqttMsg) throws Exception {
                // 接收到消息的Topic规则为：
                // /${string:domain}/${string:node-id}/${string:pattern-type}/${long:message-id}/${string:sender-node-id}/${string:tag}
                final String[] segments = StringKit.split(topic, "/");
                // Topic分段长度
                if (segments.length < 5) {
                    LOGGER.error("Invalid segments of topic: {}", topic);
                    return;
                }
                // 检查消息ID
                final String messageId = segments[3];
                if (!isDigits(messageId)) {
                    LOGGER.error("Invalid messageId segment({}) of topic: {}", messageId, topic);
                    return;
                }
                final String domain = segments[0];
                final String nodeId = segments[1];
                final String type = segments[2];
                final String senderNodeId = segments[4];
                final String tag = segments.length >= 6 ? segments[5] : "";
                final Message receivedMessage = new Message.Builder()
                        .domain(domain)
                        .targetNodeId(nodeId)
                        .type(type)
                        .messageId(Long.parseLong(messageId))
                        .senderNodeId(senderNodeId)
                        .tag(tag)
                        .payload(mqttMsg.getPayload())
                        .qos(mqttMsg.getQos())
                        .retained(mqttMsg.isRetained())
                        .build();
                if (LOGGER.isDebugEnabled()) {
                    LOGGER.debug("Received topic: {}, message: {}", topic, receivedMessage);
                }
                for (Pair<LocalFilter, MessageHandler> pair : mArgsMap.values()) {
                    if (pair.first.test(receivedMessage)) {
                        pair.second.onMessage(mSocket, receivedMessage);
                    }
                }
            }
        });
    }

    public long addHandler(LocalFilter filter, MessageHandler handler) {
        long id = mRegSerialId.nextId();
        mArgsMap.put(id, Pair.of(filter, handler));
        return id;
    }

    public void removeHandler(long id) {
        mArgsMap.remove(id);
    }

    private static boolean isDigits(String str) {
        for (char c : str.toCharArray()) {
            if (c < '0' || c > '9') {
                if (c == '-') {
                    continue;
                }
                return false;
            }
        }
        return true;
    }
}
